package com.offcn.spark.p3

import org.apache.hadoop.io.{IntWritable, Text}
import org.apache.hadoop.mapreduce.lib.output.{FileOutputFormat, TextOutputFormat}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @Auther: BigData-LGW
 * @ClassName: Action
 * @Date: 2020/12/7 22:07
 * @功能描述: $FunctionDescription
 * @Version:1.0
 */
object Action {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf()
            .setAppName("_05ActionOps")
            .setMaster("local[*]")
        val sc = new SparkContext(conf)
        val array = sc.parallelize(Array(
            "hello you",
            "hello me",
            "hello you",
            "hello you",
            "hello me",
            "hello you"
        ), 2)
        val pairs = array.flatMap(_.split("\\s+")).map((_,1))
        val ret = pairs.aggregateByKey(0)(_+_,_+_)
        //学习action操作
        /*
            collect的作用就是将分散在executor上面的rdd-partition数据，通过网络拉回到driver之上，返回值是一个数组
            尽量避免对全量数据集进行collect操作，搞不好，driver要崩溃
            所以，在collect之前先filter一部分数据，或者使用其他算子，比如take
         */
        var array1 = ret.collect()
        println("--collect---" + array1.mkString("[", ", ", "]"))
        /*
            take(N)就是从rdd中zhi只拉去其中的N条记录，返回值是一个数组
            如果rdd中的数据都是有序的，take(N) ---> topN
            take(N)的一个特例就是first，获取第一条记录
      */
        array1 = ret.take(1)
        println("--take---" + array1.mkString("[", ", ", "]"))
        /*
            reduce， reduce是一个action操作，返回值是单列内容，就相当于scala集合
            reduceByKey是一个transformation操作，返回值是一个RDD
         */
        val count = ret.values.reduce(_+_)
        println("总单词的个数：" + count)
        /**
         * countByKey,统计key出现的次数
         *  返回值是一个Map[K, Long]
         */
        val wordcount = pairs.countByKey()
        for((word,count) <- wordcount){
            println(s"word: ${word}, count: ${count}")
        }
        ret.saveAsNewAPIHadoopFile(
            path = "file:/D:/tmp/output/spark/hadoop",
            keyClass = classOf[Text],
            valueClass = classOf[IntWritable],
            outputFormatClass = classOf[TextOutputFormat[Text,IntWritable]]
        )

    }
}
